[FLINK-37443] Add returns() method to DataStream V2 API for specifying output types with lambda expressions by nilmadhab · Pull Request #26284 · apache/flink

nilmadhab · 2025-03-11T16:03:37Z

What is the purpose of the change

(For example: This pull request makes task deployment go through the blob server, rather than through RPC. That way we avoid re-transferring them on each deployment (during recovery).)

Brief change log

(for example:)

The TaskInfo is stored in the blob store on job creation time as a persistent artifact
Deployments RPC transmits only the blob storage reference
TaskManagers retrieve the TaskInfo from the blob cache

Verifying this change

Please make sure both new and modified tests in this PR follow the conventions for tests defined in our code quality guide.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment with large payloads (100MB)
Extended integration test for recovery after master (JobManager) failure
Added test that validates that TaskInfo is transferred only once across recoveries
Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2025-03-11T16:16:58Z

CI report:

eab4d02 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

…nd using .returns method

nilmadhab · 2025-03-14T04:27:35Z

@flinkbot run azure

nilmadhab · 2025-03-19T05:17:19Z

@reswqa can you please review it?

reswqa · 2025-03-19T07:15:35Z

cc @codenohup

codenohup · 2025-03-20T03:03:48Z

flink-datastream-api/pom.xml

-	</dependencies>
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-core</artifactId>


The API modules should not depend on non-API modules, because this would require users to depend on the flink-core module when developing DataStream V2 programs, but flink-core should only be used at application runnning.

You might consider moving TypeInformation to the API module, such as core-api. However, I understand that this is challenging due to its dependencies on numerous other classes.

I can copy it, but will it be okay to have these duplicate code in codebase? Or is there any other better way to do that?

The dependency of flink-core module through usingTypeInformation is only used in the interfaces located in flink-datastream-api, as the implementations are defined in flink-datastream module which are already using flink-core module.

I think it does not make much sense to copy a large number of files and maintain it just for a small feature and that too only for defining interfaces. I am open to other ideas.

The dependency of flink-core module through usingTypeInformation is only used in the interfaces located in flink-datastream-api, as the implementations are defined in flink-datastream module which are already using flink-core module.

I think it does not make much sense to copy a large number of files and maintain it just for a small feature and that too only for defining interfaces. I am open to other ideas.

Sorry for the delayed reply. I think we can introduce a class, such as TypeHint, at the API level that allows users to define a class with generics. Subsequently, this TypeHint can be converted into TypeInformation at the implementation level.
During this process, you can leverage TypeDescriptor to describe common types within the flink-datastream-api. Additionally, it would be advantageous to incorporate the tuple type into TypeDescriptor.

codenohup · 2025-03-20T03:06:33Z

...stream-api/src/main/java/org/apache/flink/datastream/api/stream/NonKeyedPartitionStream.java

+     * @param typeInfo type information as a return type hint
+     * @return This operator with a given return type hint.
+     */
+    NonKeyedPartitionStream<T> returns(TypeInformation<T> typeInfo);


KeyedPartitionStream and GlobalStream also need returns method.

okay I will add it,

okay I will add it,

BroadcastStream also need returns method

Can it be just part of datastream interface? Then it will extend all other streams by default.

Can it be just part of datastream interface? Then it will extend all other streams by default.

Thank you for the reminder. You can add the withReturnType method to the ProcessConfigurable interface, as it proves to be useful in most scenarios. Regarding situations with two output streams, users can obtain both streams individually and then apply the withReturnType method to each. Here's an example to illustrate:

NonKeyedPartitionStream.ProcessConfigurableAndTwoNonKeyedPartitionStream<Integer, String> twoOutputStream = ...; twoOutputStream.getFirst().withReturnType(TypeDescriptor.INT).process(...); twoOutputStream.getSecond().withReturnType(TypeDescriptor.String).process(...);

codenohup · 2025-03-20T03:07:34Z

...g/src/main/java/org/apache/flink/streaming/examples/dsv2/wordcount/WordCountUsingLambda.java

+ * DataStream API. It is currently in the experimental stage and is not fully available for
+ * production.
+ */
+public class WordCountUsingLambda {


Maybe you should rewrite the previous example to use lambda expressions instead of rewriting an almost identical example.

yes, I was thinking the same !

codenohup

Hi, @nilmadhab
Thank you for your contribution! I've left some comments, PTAL.

…rocessConfigurableAndKeyedPartitionStreamImpl

nilmadhab · 2025-03-27T07:18:20Z

Any other examples I can modify or test case to add? @codenohup

nilmadhab · 2025-03-27T09:31:44Z

@flinkbot run azure

codenohup · 2025-03-31T07:38:08Z

Any other examples I can modify or test case to add? @codenohup

It is best to modify the original example/test case. If it does not meet your needs, you can also add a new example/test case.

Trying out using lambda function in dataStreamV2 tests

31c909b

nilmadhab changed the title ~~Trying out using lambda function in dataStreamV2 tests~~ FLINK-37443 Trying out using lambda function in dataStreamV2 tests Mar 11, 2025

flinkbot added the component=API/DataStream label Mar 11, 2025

Nilmadhab Mondal added 3 commits March 11, 2025 17:45

Working in my sample app

e192f97

Adding requireNonNull and fix formatting

99727be

reverting test changes

b2f1f57

nilmadhab marked this pull request as ready for review March 13, 2025 20:34

Adding working example with OneInputStreamProcessFunction as lamdba a…

f9a35c4

…nd using .returns method

nilmadhab changed the title ~~FLINK-37443 Trying out using lambda function in dataStreamV2 tests~~ [FLINK-37443] Add returns() method to DataStream V2 API for specifying output types with lambda expressions Mar 13, 2025

mvn spotless:apply done

8f03b7c

codenohup reviewed Mar 20, 2025

View reviewed changes

Nilmadhab Mondal added 3 commits March 27, 2025 07:43

Adding returns method in GlobalStream and KeyedPartitionStream

97d605c

Adding returns method in ProcessConfigurableAndGlobalStreamImpl and P…

88949a1

…rocessConfigurableAndKeyedPartitionStreamImpl

mvn spotless:apply

082df16

Using lambda methods leveraging the returns method

eab4d02

github-actions bot added community-reviewed PR has been reviewed by the community. and removed component=API/DataStream community-reviewed PR has been reviewed by the community. labels Jun 30, 2025

github-actions bot added community-reviewed PR has been reviewed by the community. and removed community-reviewed PR has been reviewed by the community. labels Jul 1, 2025

github-actions bot added community-reviewed PR has been reviewed by the community. and removed community-reviewed PR has been reviewed by the community. labels Jul 17, 2025

github-actions bot added community-reviewed PR has been reviewed by the community. and removed community-reviewed PR has been reviewed by the community. labels Jul 24, 2025

github-actions bot added community-reviewed PR has been reviewed by the community. and removed community-reviewed PR has been reviewed by the community. labels Aug 15, 2025

github-actions bot added community-reviewed PR has been reviewed by the community. and removed community-reviewed PR has been reviewed by the community. labels Aug 26, 2025

Conversation

nilmadhab commented Mar 11, 2025

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Mar 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

nilmadhab commented Mar 14, 2025

Uh oh!

nilmadhab commented Mar 19, 2025

Uh oh!

reswqa commented Mar 19, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codenohup left a comment

Choose a reason for hiding this comment

Uh oh!

nilmadhab commented Mar 27, 2025

Uh oh!

nilmadhab commented Mar 27, 2025

Uh oh!

codenohup commented Mar 31, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

flinkbot commented Mar 11, 2025 •

edited

Loading